package cn._51doit.live.udfs;

import cn._51doit.live.pojo.DataBean;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava30.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava30.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 根据用户的设备ID，判断是否是用户
 */
public class IsNewUserFunctionV3 extends KeyedProcessFunction<String, DataBean, DataBean> {

    private transient ValueState<Byte> valueState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Byte> stateDescriptor = new ValueStateDescriptor<>("is-exist-state", Byte.class);
        valueState = getRuntimeContext().getState(stateDescriptor);
    }

    @Override
    public void processElement(DataBean bean, Context ctx, Collector<DataBean> out) throws Exception {

        String deviceId = bean.getDeviceId();

        Byte exist = valueState.value();
        if (exist == null) {
            bean.setIsN(1);
            exist = 1;
            valueState.update(exist);
        }

        out.collect(bean);
    }
}
